package org.heapbd.evaluation.impala;

import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
import java.util.Properties;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.heapbd.evaluation.utility.Constants;

public class ImpalaSpecificQueryTest extends Configured implements Runnable {
	
	public static void main(String[] args)
	{
		Properties properties = System.getProperties();
		String[] propKeys = properties.keySet().toArray(new String[properties.size()]);
		Arrays.sort(propKeys);
		for (String propKey : propKeys)
		{
			System.out.println(propKey + "=" + properties.getProperty(propKey));
		}
	}
	
	private long inputRecordCount;
	private String[] specificFilters;
	private int threadId;
	private Properties properties;
	
	public void setup(long inputRecordCount, String[] specificFilters, int threadId, 
			Properties properties)
	{
		this.inputRecordCount 	= inputRecordCount;
		this.specificFilters 	= specificFilters;
		this.threadId 			= threadId;
		this.properties 		= properties;
	}
	
	public void run()
	{
		// let's give the example of a MySQL implementation.
		String driver 	= properties.getProperty(Constants.DATABASE_DRIVER);
		String url 		= properties.getProperty(Constants.DATABASE_URL);
		String username = properties.getProperty(Constants.DATABASE_USER);
		String password = properties.getProperty(Constants.DATABASE_PASS);
		String hdfsDir 	= properties.getProperty(Constants.HDFS_DIRECTORY);
		
		try {
			Class.forName(driver);
			Connection conn = DriverManager.getConnection(url, username, password);
			Statement stmt = conn.createStatement();
			
			FileSystem fs = FileSystem.get(getConf());
			String filterTable = loadFiltersToTable(stmt, fs, hdfsDir);
			
			String resultTable = runQuery(stmt, filterTable);
			
			cleanUp(stmt, resultTable, filterTable);
			
			stmt.close();
			conn.close();
		} catch (Exception e) {
			System.err.println("Specific Query Test " + threadId + " failed:");
			e.printStackTrace();
		}
	}
	
	protected void cleanUp(Statement stmt, String resultTable, String filterTable)
			throws SQLException
	{
		String cleanup = properties.getProperty(Constants.CLEANUP);
		if (cleanup != null && "true".equals(cleanup))
		{
			stmt.execute("drop table " + resultTable);
			stmt.execute("drop table " + filterTable);
		}
	}
	
	protected String runQuery(Statement stmt, String filterTable) throws SQLException
	{
		String resultTable = "SPECIFIC_FILTER_RESULTS_" +
				inputRecordCount + "_" + threadId + "_" + specificFilters.length;
		String dataTable = "DATA_" + inputRecordCount + "_" + threadId;
		
		stmt.execute("create table " + resultTable + " as " +
				"select d.* " +
				"from " + dataTable + " d " +
				"join " + filterTable + "f on f.FILTER = d.CLIENT_IP ");
		
		return resultTable;
	}
	
	protected String loadFiltersToTable(Statement stmt, FileSystem fs, 
			String hdfsDir) throws SQLException, IOException
	{
		int specificFilterCount = specificFilters.length;
		
		// copy the filters in a file to HDFS
		String tempDir = System.getProperty("java.io.tmpdir");
		String filterFile = tempDir + File.separator + "filters.txt";
		OutputStream out = new BufferedOutputStream(new FileOutputStream(filterFile));
		for (String filter : specificFilters)
		{
			out.write((filter + "\n").getBytes("UTF-8"));
		}
		out.close();
		fs.copyFromLocalFile(new Path(filterFile), new Path(hdfsDir + "/filters_" 
				+ inputRecordCount + "_" + threadId + "_" + specificFilterCount + ".txt"));
		
		// load filters first
		String filterTable = "SPECIFIC_FILTER_FILTERS_" + 
				inputRecordCount + "_" + threadId + "_" + specificFilterCount;
		stmt.execute("create table " + filterTable + " ( " +
				"  FILTER varchar(50) " +
				") location '" + hdfsDir + "' ");
		
		return filterTable;
	}
}
